package com.sparrow.rpc.socket;

import com.sparrow.common.thread.ThreadPool;
import com.sparrow.common.util.AtomicUtil;
import com.sparrow.common.util.DynamicProxyUtil;
import com.sparrow.common.util.PropertiesUtil;
import com.sparrow.rpc.service.RpcService;
import com.sparrow.rpc.future.RpcFuture;
import com.sparrow.rpc.future.SettableRpcFuture;
import com.sparrow.rpc.socket.codec.RpcDecoder;
import com.sparrow.rpc.socket.codec.RpcEncoder;
import com.sparrow.rpc.socket.codec.RpcLengthFieldBasedFrameDecoder;
import com.sparrow.rpc.socket.handler.RpcClientHandler;
import com.sparrow.rpc.socket.handler.RpcClientHeatBeatHandler;
import com.sparrow.rpc.socket.msg.req.RpcLoginRequest;
import com.sparrow.rpc.socket.msg.req.RpcRequest;
import com.sparrow.rpc.socket.session.RpcSession;
import com.sparrow.rpc.util.RpcUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

/**
 * RpcClient 类用于创建 RPC 客户端，通过 Netty 框架实现与服务端的通信。
 * 该类实现了 InvocationHandler 接口，可用于创建代理对象。
 */
public class RpcClient extends ChannelInitializer<Channel> implements InvocationHandler {

    // 日志记录器
    private static final Logger logger = LoggerFactory.getLogger(RpcClient.class);

    // 用于存储接口类和对应的代理对象
    private final ConcurrentHashMap<Class<?>, Object> proxyMap = new ConcurrentHashMap<>();

    // 处理连接和 IO 读写的线程组，所有客户端复用
    public static final EventLoopGroup group = new NioEventLoopGroup();

    // Netty 客户端启动器
    private Bootstrap bootstrap = new Bootstrap();

    // 服务端 IP 地址
    private final String ip;
    // 服务端端口号
    private final int port;
    /**
     * 节点Id
     */
    private final String nodeId;

    private AtomicReference<RpcSession> sessionRef = new AtomicReference<>(null);

    private int MAX_RETRIES = 1200;

    private AtomicBoolean connectingFlag = new AtomicBoolean(false);
    /**
     * 缓存未连接前的消息，连接成功会将该缓存的消息发送出去
     */
    private Queue<Object> waitSendQueue = new LinkedBlockingQueue<>(1000);

    private volatile boolean stop;

    private Timeout reconnectTimeout;

    /**
     * 构造函数，初始化服务端的 IP 地址和端口号。
     *
     * @param ip   服务端 IP 地址
     * @param port 服务端端口号
     */
    public RpcClient(String nodeId, String ip, int port) {
        this.nodeId = nodeId;
        this.ip = ip;
        this.port = port;

        bootstrap.group(group).channel(NioSocketChannel.class)
        .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
        bootstrap.handler(this);
    }

    public String getNodeId() {
        return nodeId;
    }

    public void connect(ChannelFutureListener listener) {
        if (!connectingFlag.compareAndSet(false, true)) {
            return;
        }
        if (stop) {
            return;
        }
        connect(listener, 0);
    }

    private void connect(ChannelFutureListener listener, int retries) {
        if (retries >= MAX_RETRIES) {
            connectingFlag.set(false);
            logger.error("达到最大重试次数，无法连接到服务器 nodeId:{} ip:{} port:{}", nodeId, ip, port);
            return;
        }
        if (stop) {
            return;
        }
        try {
            // 尝试异步连接，避免阻塞当前线程
            ChannelFuture future = bootstrap.connect(ip, port);
            future.addListener(new ConnectionListener(listener, retries));
        } catch (Exception e) {
            logger.error("rpc连接失败，节点id:{} ip:{} port:{} ", nodeId, ip, port, e);
        }
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        String methodKey = RpcUtil.getMethodKey(method);
        RpcRequest request = new RpcRequest();
        request.setUuid(RpcUtil.nextUUID());
        request.setMethodKey(methodKey);
        request.setArgs(args);

        Class<?> returnType = method.getReturnType();
        if (RpcUtil.isVoidMethod(returnType)) {// 接口返回 void 就不需要future了
            send(request);
            return null;
        }

        RpcFuture<?> future;
        if (SettableRpcFuture.class.isAssignableFrom(returnType)) {
            future = SettableRpcFuture.of();
        } else {
            future = new RpcFuture<>();
        }
        future.setRequestId(request.getUuid());
        future.setMethodInfo(methodKey);
        RpcService.addFuture(future);
        send(request);
        return future;
    }

    public boolean isActive() {
        RpcSession session = sessionRef.get();
        return session != null && session.isActive();
    }

    private class ConnectionListener implements ChannelFutureListener {
        private final ChannelFutureListener afterConnectedListener;
        private final int retries;

        public ConnectionListener(ChannelFutureListener listener, int retries) {
            this.afterConnectedListener = listener;
            this.retries = retries;
        }

        @Override
        public void operationComplete(ChannelFuture ff) throws Exception {
            if (ff.isSuccess()) {
                bindSession(ff);
                afterConnectedListener.operationComplete(ff);
            } else {
                logger.error("rpc 连接失败 nodeId:{} ip:{} port:{}，重试次数: {}", nodeId, ip, port, retries + 1);
                RpcClient.this.reconnectTimeout = ThreadPool.TIMER.newTimeout(timeout -> {
                    connect(afterConnectedListener, retries + 1);
                }, 3000, TimeUnit.MILLISECONDS);
            }
        }
    }

    private synchronized void bindSession(ChannelFuture ff) {
        RpcSession session = new RpcSession();
        session.setIdentity(nodeId);
        session.setChannel(ff.channel());
        if (sessionRef.compareAndSet(null, session)) {
            connectingFlag.set(false);
            logger.info("RPC Client Connected to server {}:{}", ip, port);
        }
    }


    /**
     * 向服务端发送消息。
     *
     * @param msg 要发送的消息
     */
    public void send(Object msg) {
        RpcSession session = sessionRef.get();
        if (session == null) { //sessionRef 原则设置成功时 waitSendQueue会发送完
            if (!waitSendQueue.offer(msg)) {
                logger.warn("消息队列已满，丢弃消息: {}", msg);
                return;
            }
            if (!connectingFlag.get()) {
                connect();
            }
        } else {
            session.writeAndFlush(msg);
        }
    }

    /**
     * 并发情况下 只有一个线程能真正发起连接 使用原子保证
     */
    public void connect() {
        connect(f -> {
            RpcSession session = sessionRef.get();
            //里面发送登录请求
            RpcLoginRequest regReq = new RpcLoginRequest(PropertiesUtil.getProperty("rpc.node.id"), PropertiesUtil.getProperty("rpc.node.ip"), PropertiesUtil.getPropertyAsInt("rpc.node.port"));
            session.writeAndFlush(regReq);

            Object task;
            while ((task = waitSendQueue.poll()) != null) {
                try {
                    session.writeAndFlush(task);
                } catch (Exception e) {
                    logger.error("Failed to send task", e);
                }
            }
        });
    }


    /**
     * 获取指定接口的代理对象。
     *
     * @param clazz 接口类
     * @param <T>   接口类型
     * @return 接口的代理对象
     */
    public <T> T getProxy(Class<T> clazz) {
        Object proxy;
        if (clazz.isInterface()) {
            proxy = proxyMap.computeIfAbsent(
                    clazz,
                    k -> DynamicProxyUtil.createJDKProxy(clazz, this)
            );
        } else { //不是接口的话使用cglib生成动态代理
            proxy = proxyMap.computeIfAbsent(
                    clazz,
                    k -> (T) DynamicProxyUtil.createCGLIBProxy(clazz, (o, method, args, methodProxy) -> RpcClient.this.invoke(null, method, args)));
        } return (T) proxy;
    }


    /**
     * 关闭 RpcClient 并释放资源。
     */
    public void shutdown() {
        this.stop = true;
        RpcSession session = sessionRef.get();
        sessionRef.set(null);
        if (this.reconnectTimeout != null) {
            this.reconnectTimeout.cancel();
            this.reconnectTimeout = null;
        }
        if (session != null) {
            try {
                session.getChannel().close().sync();
            } catch (InterruptedException e) {
                logger.error("", e);
            }
        }
    }

    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        // 添加编码器
        pipeline.addLast(new RpcEncoder());
        // 添加基于长度字段的帧解码器
        pipeline.addLast(new RpcLengthFieldBasedFrameDecoder(Integer.MAX_VALUE));
        // 添加解码器
        pipeline.addLast(new RpcDecoder());
        //每10秒写心跳
        pipeline.addLast("idleStateHandler", new IdleStateHandler(0, 10, 0));
        pipeline.addLast("tcpIdleHandler", new RpcClientHeatBeatHandler());
        // 添加客户端处理器
        pipeline.addLast(RpcClientHandler.getInstance());
    }

    public void closeSession() {
        AtomicUtil.spinSet(sessionRef, null);
        logger.info("rpc client close session");
    }
}